package com.bw.ad1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.MyKafkaUtil;
import com.bw.bean.CitySort;
import com.bw.bean.OrgSort;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.TimeZone;

public class Test12 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        DataStreamSource<String> stream = env.addSource(MyKafkaUtil.getKafkaConsumer("tms_ods_yk9", "test4" + System.currentTimeMillis()));

        //筛选订单数据
        SingleOutputStreamOperator<String> orderInfoDS = stream.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                JSONObject jsonObject = JSON.parseObject(s);
                String tableName = jsonObject.getString("table");
                if ("order_org_bound".equals(tableName)) {
                    String status = jsonObject.getJSONObject("data").getString("status");
                    if ("64001".equals(status) || "64004".equals(status)) {
                        return true;
                    }
                    return true;
//                    return false;
                }
                return false;
            }
        });
//        orderInfoDS.print();

        // 变成JSON数据
        SingleOutputStreamOperator<JSONObject> mapStream = orderInfoDS.map(new MapFunction<String, JSONObject>() {
            @Override
            public JSONObject map(String s) throws Exception {
                return JSON.parseObject(s);
            }
        });
        // 添加水位线
        SingleOutputStreamOperator<JSONObject> watermarkStream = mapStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                // 时间戳13位
                return element.getJSONObject("data").getLong("create_time") ;
            }
        }));

//        watermarkStream.print();


//        // 每个机构分拣平常时长
        SingleOutputStreamOperator<OrgSort> processStream = watermarkStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("data").getString("org_id");
            }
        }).window(TumblingEventTimeWindows.of(Time.minutes(60))).process(new ProcessWindowFunction<JSONObject, OrgSort, String, TimeWindow>() {
            private MapState<String, String> mapState;

            @Override
            public void open(Configuration parameters) throws Exception {
                mapState = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("map_state", String.class, String.class));
            }

            @Override
            public void process(String key, ProcessWindowFunction<JSONObject, OrgSort, String, TimeWindow>.Context context, Iterable<JSONObject> iterable, Collector<OrgSort> collector) throws Exception {
//               /**
//                *
//                *   org_id   order_id   in   out
//                *    1          10      100        insert
//                *    1          10      100  200   update
//                *    1          11      110        insert
//                *    1          11      110  300   update
//                *
//                *
//                ** /


                int all_diff = 0;
                int count = 0;
                Iterator<JSONObject> iterator = iterable.iterator();
                while (iterator.hasNext()) {
                    JSONObject orderInfo = iterator.next();
                    // 获取订单ID
                    String ID = orderInfo.getJSONObject("data").getString("order_id");

                    // 入库时间
                    Long in_time = orderInfo.getJSONObject("data").getLong("inbound_time");
                    // 入库时间
                    Long out_time = orderInfo.getJSONObject("data").getLong("outbound_time");

                    if (out_time != null){
                        // 取出时间
                        String value = mapState.get(ID);


                        if (value != null) {
                            // 放下单的时间
                            mapState.put(ID, in_time + "");
                        } else {
                            // 分拣时长
                            long diff = out_time - in_time;
                            all_diff += diff;
                            count++;

                        }

                        // 平均分拣时长
                        int avg_ts =  all_diff / count;

                        OrgSort orgSort = new OrgSort();
                        orgSort.setOrgId(key);
                        orgSort.setAvgTs(avg_ts);
                        orgSort.setEdt(context.window().getEnd());
                        collector.collect(orgSort);
                    }

                }
            }
        });
////        processStream.print("processStream>>");
//        /*
//            city_id  avg_ts
//                1     20
//                2     30
//                3     5
//                4     80
//
//
//
//                1   10
//                2   30
//                5   60
//         */
//        // 根据窗口在进行keyby
        SingleOutputStreamOperator<String> triggerTSStream = processStream.keyBy(new KeySelector<OrgSort, String>() {
            @Override
            public String getKey(OrgSort orgSort) throws Exception {
                return orgSort.getEdt() + "";
            }
        }).process(new KeyedProcessFunction<String, OrgSort, String>() {
            private ListState<OrgSort> orgSortListState;
            private ValueState<Long> triggerTS;

            @Override
            public void open(Configuration parameters) throws Exception {
                triggerTS = getRuntimeContext().getState(new ValueStateDescriptor<Long>("triggerTS", Long.class));
                orgSortListState = getRuntimeContext()
                        .getListState(new ListStateDescriptor<OrgSort>("hotItems", OrgSort.class));
            }

            @Override
            public void processElement(OrgSort orgSort, KeyedProcessFunction<String, OrgSort, String>.Context context, Collector<String> collector) throws Exception {
                orgSortListState.add(orgSort);
                if (triggerTS.value() == null) {
                    // 窗口结束时间加1秒
                    context.timerService().registerEventTimeTimer(orgSort.getEdt() + 1L);
                    triggerTS.update(orgSort.getEdt());
                }
            }

            // 定时器触发方法
            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, OrgSort, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                // 拿出状态里面的所有数据
                Iterable<OrgSort> cs = this.orgSortListState.get();

                // 存储最终的结果
                ArrayList<OrgSort> result = new ArrayList<>();


                for (OrgSort orgSort : cs) {
                    result.add(orgSort);
                }
                // 清空状态
                this.orgSortListState.clear();
                // 清空定时器
                triggerTS.clear();

                // 对result
                result.sort((o1, o2) -> o1.getAvgTs().intValue() - o2.getAvgTs().intValue());


                // 拼出最后的显示结果
                StringBuilder sb = new StringBuilder();
                sb.append("窗口结束时间: " + (timestamp - 1) + "\n");
                sb.append("---------------------------------\n");
                for (int i = 0; i < result.size(); i++) {
//
                    sb.append("第"+(i+1)+"名，中转站"+result.get(i).getOrgId()+"健康评分:"+result.get(i).getScore() + "\n");
                }
                sb.append("---------------------------------\n\n");
                out.collect(sb.toString());

            }
        });
//
        triggerTSStream.print("processStream>>");

        env.execute();
    }
}
